package jms;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.logging.Level;
import java.util.logging.Logger;

import javax.jms.Message;
import javax.jms.MessageListener;


public class MessageDispatcherManager implements MessageListener {
	private ExecutorService servicePool;
	private final AtomicInteger serviceNumber = new AtomicInteger();
	private final BlockingQueue<Message> receivedMessages = new LinkedBlockingQueue<Message>(100);
	private final ConcurrentHashMap<String, MessageDispatcher> dispatcherMap = 
			new ConcurrentHashMap<String, MessageDispatcher>();
	
	public void start() {
		//servicePool = Executors.newCachedThreadPool();
		int poolSize = 2;
		servicePool = Executors.newFixedThreadPool(poolSize, new NamedThreadFactory("MsgProcess"));
		for(int k=0; k<poolSize; ++k) {
			servicePool.execute(new Runnable() {
				
				public void run() {
					while(true) {
						final Message rMsg;
						
						try {
							rMsg = receivedMessages.take();
						}catch(InterruptedException ex) {
							break;
						}
						if(servicePool.isShutdown())
							break;
						if(rMsg == null)
							continue;
						
						try {
							processMessage(rMsg);
						} catch (Exception ex) {
							Logger.getAnonymousLogger().log(Level.SEVERE, "", ex);
						}
					}
				}
			});
		}
		dispatcherMap.clear();
	}
	
	public void stop() {
		servicePool.shutdown();
		
		try {
			servicePool.awaitTermination(10, TimeUnit.SECONDS);				
		}catch(InterruptedException ex){
			Logger.getAnonymousLogger().warning("Interrupted");
		}
		servicePool.shutdownNow();
		servicePool = null;
		
		dispatcherMap.clear();
		
	}
	
	public void register(String correlationId, MessageDispatcher dispatcher) {
		dispatcherMap.put(correlationId, dispatcher);
	}
	
	public void onMessage(Message message) {
		//processMessageInMultiThreads(message);
		deliverToQueue(message);
	}
	
	void deliverToQueue(Message message) {
		try {
			receivedMessages.put(message);
		} catch(InterruptedException ex) {
			//simply return
			return;
		}
	}
	
	void processMessageInMultiThreads(final Message message) {
		while(true) {
			int curr = serviceNumber.incrementAndGet();
			if(curr >= 10) {
				Logger.getAnonymousLogger().warning("Exceed max process thread + " + curr);
				serviceNumber.decrementAndGet();
				try {
					Thread.sleep(100);	//100 milliseconds, disorder risks
				} catch (InterruptedException ex) {
					Logger.getAnonymousLogger().warning("Interrupted");
					return;
				}
			}
			else
				break;	//pass through
		}
		this.servicePool.execute(new Runnable() {
			public void run() {
				try {
					processMessage(message);
				} catch (Exception ex) {
					Logger.getAnonymousLogger().log(Level.SEVERE, "", ex);
				} finally {
					serviceNumber.decrementAndGet();
				}
			}
		});
	}
	
	void processMessage(Message message) throws Exception {
		String cId = message.getJMSCorrelationID();
		if(cId == null) {
			Logger.getAnonymousLogger().severe("Unexpected cid");
			return;
		}
		MessageDispatcher dispatcher = dispatcherMap.get(cId);
		if(dispatcher == null) {
			Logger.getAnonymousLogger().warning("Missing dispatcher for => " + cId);
			return;
		}
		String isSingle = dispatcher.getProperty("Single");
		if(isSingle.equals(Boolean.TRUE.toString()))
			dispatcherMap.remove(cId);
		
		//System.out.println("Message received => " + cId);
		dispatcher.onDispatch(message);
	}
}